package Utils.Concurrent;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * TaskPool is an asynchronous task manager used to handle a large number of
 * simultaneous tasks at once or just for simplified control over multiple
 * threads.
 * 
 * Encapsulates the java concurrent utility ExecutorService
 * 
 * @author micha
 * 
 */

public class TaskPool<T> {

	/**
	 * the encapsulated executor
	 */
	private ExecutorService executor;

	/**
	 * stores pairs of tasks and it's corresponding future
	 */
	private Map<AsyncTask<T>, Future<T>> tasks;

	/**
	 * stores the pool maximum tasks running at once
	 */
	private int poolSize;

	/*
	 * -------------------------------------------------------------------
	 * --------------------------- Constructors --------------------------
	 * -------------------------------------------------------------------
	 */

	/**
	 * Creates a thread pool that reuses a fixed number of threads operating off
	 * a shared unbounded queue
	 * 
	 * @param poolSize
	 *            number of tasks in pool that can be invoked at once
	 */
	public TaskPool(int poolSize) {
		executor = Executors.newFixedThreadPool(poolSize);
		tasks = new HashMap<AsyncTask<T>, Future<T>>();
		this.poolSize = poolSize;
	}

	/**
	 * Creates a thread pool that reuses a fixed number of threads operating off
	 * a shared unbounded queue
	 * 
	 * @param poolSize
	 *            number of tasks in pool that can be invoked at once
	 * 
	 * @param executeNow
	 *            whether or not to execute the task immediately (if tasks are
	 *            not executed immediately, they are stored for future use to be
	 *            called by the executeAll() method )
	 * 
	 * @param tasks
	 *            tasks to be executed (or stored)
	 */
	public TaskPool(int poolSize, boolean executeNow, AsyncTask<T>... tasks) {
		executor = Executors.newFixedThreadPool(poolSize);
		this.tasks = new HashMap<AsyncTask<T>, Future<T>>();
		Future<T> future = null;

		for (AsyncTask<T> task : tasks) {
			if (executeNow) {
				// execute immediately
				future = executor.submit(task);
				this.tasks.put(task, future);
			} else {
				// store for future use , define task future to be null
				this.tasks.put(task, null);
			}
		}

		this.poolSize = poolSize;
	}

	/*
	 * -------------------------------------------------------------------
	 * ----------------------- pool API functions ------------------------
	 * -------------------------------------------------------------------
	 */

	/**
	 * adds a new task to the pool
	 * 
	 * @param executeNow
	 *            whether or not to execute the task immediately
	 * 
	 * @param task
	 *            task to be executed
	 */
	public void addTaskToPool(boolean executeNow, AsyncTask<T> task) {
		try {
			if (executeNow) {
				// execute immediately
				Future<T> future = executor.submit(task);
				this.tasks.put(task, future);
			} else {
				// store for future use, define task future to be null
				this.tasks.put(task, null);
			}

		} catch (RejectedExecutionException e) {
			// abort task submission
		}
	}

	/**
	 * pushes a task to the pool, stops all actively executing tasks, halts the
	 * processing of waiting tasks and executes an urgent task in a new
	 * initialized pool.
	 * 
	 * 
	 * @param urgentTask
	 *            urgent task to be executed instead off all previous tasks
	 */
	public void pushUrgentTask(AsyncTask<T> urgentTask) {
		try {

			// shuts down all other current active and pending tasks
			executor.shutdownNow();

			// Initialize pool
			executor = Executors.newFixedThreadPool(poolSize);

			// initialize result set
			clean();

			// execute urgent task
			Future<T> future = executor.submit(urgentTask);
			this.tasks.put(urgentTask, future);

		} catch (Exception e) {
			System.out.println(e);
		}
	}

	/**
	 * invokes all stored tasks, and stores each task corresponding future
	 */
	public void executeAll() {
		Future<T> future = null;
		Iterator<Entry<AsyncTask<T>, Future<T>>> taskIterator = tasks
				.entrySet().iterator();
		Map.Entry<AsyncTask<T>, Future<T>> taskPair;

		while (taskIterator.hasNext()) {
			taskPair = taskIterator.next();
			// execute task
			future = executor.submit(taskPair.getKey());

			// store task future
			taskPair.setValue(future);
			//taskIterator.remove(); // avoids a ConcurrentModificationException
		}

	}

	/**
	 * initiates the task pairs
	 * 
	 * note: this is a dangerous function and should not be used lightly, it
	 * should be used only in case all tasks are completed and all results are
	 * already used.
	 */
	private void clean() {
		this.tasks = new HashMap<AsyncTask<T>, Future<T>>();
	}

	/**
	 * initiates all tasks results
	 */
	public void cleanResults() {
		this.tasks = new HashMap<AsyncTask<T>, Future<T>>();

		Iterator<Entry<AsyncTask<T>, Future<T>>> taskIterator = tasks
				.entrySet().iterator();
		Map.Entry<AsyncTask<T>, Future<T>> taskPair;

		while (taskIterator.hasNext()) {
			taskPair = taskIterator.next();
			taskPair.setValue(null);
		}
	}

	/**
	 * initiates an orderly shutdown in which previously submitted tasks are
	 * executed, but no new tasks will be accepted.
	 */
	public void close() {
		executor.shutdown();
		clean();
	}

	/**
	 * blocks until all tasks in the subset have completed execution
	 */
	public void waitForAll(AsyncTask<T>... tasksSubset) {
		for (AsyncTask<T> task : tasksSubset) {

			if (this.tasks.containsKey(task)) {

				try {
					// wait for task to finish
					this.tasks.get(task).get();
				} catch (Exception e) {
				}
			}
		}
	}

	/**
	 * blocks until all tasks have completed execution
	 * 
	 * note: previously submitted tasks are executed, but no new tasks will be
	 * accepted while waiting
	 */
	public void waitForAll() {
		try {
			executor.shutdown();
			executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
		} catch (InterruptedException e) {
			System.out.println(e);
		}
	}

	/**
	 * waits if necessary for the computation of all tasks in pool to complete,
	 * and then retrieves the result of each task independently.
	 * 
	 * note : getResults() does not closes the TaskPool, nor initializes the
	 * result set, hence the client should use close() or cleanResults() after
	 * getting the task results.
	 */
	public List<T> getResults() {
		List<T> results = new ArrayList<T>();

		Iterator<Entry<AsyncTask<T>, Future<T>>> taskIterator = tasks
				.entrySet().iterator();
		Map.Entry<AsyncTask<T>, Future<T>> taskPair;

		try {
			while (taskIterator.hasNext()) {
				taskPair = taskIterator.next();

				results.add(taskPair.getValue().get());

			}

		} catch (Exception e) {
			e.printStackTrace();
		}
		return results;
	}

	/**
	 * waits if necessary for the computation of all tasks received and then
	 * retrieves the result of each task independently.
	 * 
	 * note : getResults() does not closes the TaskPool, nor initializes the
	 * result set, hence the client should use close() or cleanResults() after
	 * getting the task results.
	 * 
	 * @param tasksSubset
	 *            a subset of the tasks currently in the task pool (if no such
	 *            task found in the task pool, it is ignored and does not store
	 *            any value regarding this task in the result list)
	 */
	public List<T> getResults(AsyncTask<T>... tasksSubset) {
		List<T> results = new ArrayList<T>();

		try {
			for (AsyncTask<T> task : tasksSubset) {

				if (this.tasks.containsKey(task)) {
					results.add(this.tasks.get(task).get());
				}

			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return results;
	}

	/**
	 * returns false iff all tasks in pool are completed. Completion may be due
	 * to normal termination, an exception, or cancellation.
	 * 
	 */
	public boolean hasRunningTask() {

		Iterator<Entry<AsyncTask<T>, Future<T>>> taskIterator = tasks
				.entrySet().iterator();
		Map.Entry<AsyncTask<T>, Future<T>> taskPair;

		while (taskIterator.hasNext()) {
			taskPair = taskIterator.next();
			if (!taskPair.getValue().isDone()) {
				return true;
			}

		}

		return false;
	}

	/**
	 * returns false iff all tasks in task subset are completed. Completion may
	 * be due to normal termination, an exception, or cancellation.
	 * 
	 * @param tasksSubset
	 *            a subset of the tasks currently in the task pool (if no such
	 *            task found in the task pool, it is ignored)
	 */
	public boolean hasRunningTask(AsyncTask<T>... tasksSubset) {

		for (AsyncTask<T> task : tasksSubset) {
			if (this.tasks.containsKey(task) && !this.tasks.get(task).isDone()) {
				return true;
			}
		}

		return false;
	}

	/**
	 * Attempts to stop all active tasks in the task subset (only interruptible
	 * ones will succeed)
	 */
	public void killAll(AsyncTask<T>... tasksSubset) {
		for (AsyncTask<T> task : tasksSubset) {

			if (this.tasks.containsKey(task)) {

				// Attempts to cancel execution of this task. Attempt will fail
				// if the task has already completed or already been cancelled
				this.tasks.get(task).cancel(true);
			}
		}
	}

	/**
	 * Attempts to stop all active tasks (only interruptible ones will succeed)
	 * and halts the processing of waiting tasks
	 */
	public void killAll() {
		this.executor.shutdownNow();
		clean();
	}

}
